package org.myorg.datasource.demo.kafka;


import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;

import java.util.Properties;

public class KafkaDsFlinkConsumer {

	public static void main(String[] args) throws Exception {
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		Properties props = new Properties();
		props.put("bootstrap.servers", "172.18.1.87:9092");
//		props.put("zookeeper.connect", "172.18.1.87:2181");
		props.put("group.id", "metric-group");
		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		props.put("auto.offset.reset", "latest");

		DataStreamSource<String> dataStreamSource = env.addSource(
				new FlinkKafkaConsumer011<>(
						KafkaDsProducer.topic,
						new SimpleStringSchema(), // String 序列化
						props
				)
		).setParallelism(1);

		dataStreamSource.print();
		env.execute("Flink add data source");
	}
}
